/**
 * 
 */
package org.metaverse.h2;

import gnu.trove.TLongHashSet;

import java.io.IOException;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Iterator;

import javax.sql.DataSource;

import org.metaverse.h2.cache.ValueCache;
import org.metaverse.warehouse.Senary;
import org.metaverse.warehouse.Senary.Slot;
import org.singularity.io.CloseableIterator;

/**
 * @author vjache
 *
 */
public class SenaryCacheWarmIterator implements CloseableIterator<Iterator<Senary>> {

	private final Connection _conn;
	private final Statement _stmt;
	private final ResultSet _rs;
	private final ValueCache _nodeCache;
	
	private ArrayList<Senary> _page;
	private final int PAGE_SIZE = 1000;
	private final Slot[] _selectSlots;

	/**
	 * 
	 */
	public SenaryCacheWarmIterator(
			DataSource aConnPool, 
			String aSql, 
			ValueCache aNodeCache, 
			Slot...aSelect) 
		throws SQLException
	{
		_conn = aConnPool.getConnection();
		_stmt = _conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
		_rs = _stmt.executeQuery(aSql);
		_nodeCache = aNodeCache;
		_selectSlots = aSelect;
		advance();
	}
	
	private void advance()throws SQLException
	{
		final TLongHashSet set = new TLongHashSet(PAGE_SIZE*5);
		int cnt = 0;
		while(_rs.next())
		{
			final long[] row = new long[]{
					_rs.getLong(1),
					_rs.getLong(2),
					_rs.getLong(3),
					_rs.getLong(4),
					_rs.getLong(5)};
			addIdsToWarmUp(set, row);
			if(_page==null)
				_page = new ArrayList<Senary>(PAGE_SIZE);
			_page.add(new LazySenary(row,_rs.getTimestamp(6).getTime(), _nodeCache));
			
			cnt++;
			
			if(cnt>=PAGE_SIZE)
				break;
		}
		if(!set.isEmpty())
			warmUpCache(set);
		if(cnt==0)
			_page = null;
	}

	private final void addIdsToWarmUp(final TLongHashSet set, final long[] row) {
		if(_selectSlots==null || _selectSlots.length==0)
			set.addAll(row);
		else
			for(Slot slot:_selectSlots)
				switch(slot)
				{
				case Subject: set.add(row[1]);
				case Predicate: set.add(row[2]);
				case Object: set.add(row[3]);
				case Context: set.add(row[4]);
				case User: set.add(row[5]);
				}
	}

	private void warmUpCache(TLongHashSet set) {
		_nodeCache.warmUpIds(set);
	}

	/* (non-Javadoc)
	 * @see java.io.Closeable#close()
	 */
	@Override
	public void close() throws IOException {
		IOException exc = null;
		if(_rs!=null)
			try {_rs.close();} 
			catch (SQLException e) 
			{exc = new IOException(e);}
		if(_stmt!=null)
			try {_stmt.close();} 
			catch (SQLException e) 
			{exc =  new IOException(e);}
		if(_conn!=null)
			try {_conn.close();} 
			catch (SQLException e) 
			{exc =  new IOException(e);}
			
		if(exc!=null)
			throw exc;
	}

	/* (non-Javadoc)
	 * @see java.util.Iterator#hasNext()
	 */
	@Override
	public boolean hasNext() 
	{return _page!=null;}

	/* (non-Javadoc)
	 * @see java.util.Iterator#next()
	 */
	@Override
	public Iterator<Senary> next() {
		ArrayList<Senary> ret = _page;
		try {
			advance();
		} catch (SQLException e) {
			throw new RuntimeException(e);
		}
		return ret.iterator();
	}

	/* (non-Javadoc)
	 * @see java.util.Iterator#remove()
	 */
	@Override
	public void remove() {
		throw new UnsupportedOperationException();
	}

}
